#include "net/EventLoop.h"
#include "net/Pollers/EPoller.h"
#include "Base/CurrectThread.h"
#include "Base/Logger.h"
#include "net/TimerId.h"
#include <stdlib.h>
#include <unistd.h>
#include <sys/eventfd.h>
#include <errno.h>

namespace toym
{
namespace net
{


namespace detail
{

int createEventFd() {
    int fd = ::eventfd(0U, EFD_NONBLOCK|EFD_CLOEXEC);
    if (fd < 0) {
        CurrectThread::saveErrno(errno);
        LOG_ERROR << "createEventFd() failed";
        ::abort();
    }
    return fd;
}

void readEventFd(int fd) {
    int64_t i64;
    ssize_t n = ::read(fd, &i64, sizeof(i64));
    LOG_DEBUG << "readEventFd: " << fd;
    if (n < 0)
        CurrectThread::saveErrno(errno); // set errno
    if (n != 8) {
        LOG_WARNING << "readEventFd(): read " << n << " bytes instead of 8";
        CurrectThread::saveErrno(0); // reset errno
    }
}

void writeEvnetFd(int fd) {
    int64_t i64 = 1;
    ssize_t n = ::write(fd, &i64, sizeof(i64));
    LOG_DEBUG << "writeEvnetFd: " << fd;
    if (n < 0)
        CurrectThread::saveErrno(errno);
    if (n != 8) {
        LOG_WARNING<< "writeEventFd(int): write " << n << " bytes instead of 8";
        CurrectThread::saveErrno(0);
    }
}

void defaultEventErrorHandler() {
    LOG_WARNING << " error happened in event fd ";
}

} // namespace detail


////////////////////////////////////////////////////////
/////    static members
////////////////////////////////////////////////////////

thread_local EventLoop* EventLoop::loopInThisThread = nullptr;


////////////////////////////////////////////////////////
/////    Public methods
////////////////////////////////////////////////////////

EventLoop::EventLoop()
    : m_poller(new EPoller()),
      m_tid(CurrectThread::getTid()),
      m_eventChannel(detail::createEventFd(), this),
      m_functors(),
      m_functorMutex(),
      m_timerQueue(this),
      m_looping(false)
{
    if (loopInThisThread != nullptr) {
        LOG_ERROR << "Another loop at " << loopInThisThread 
                  << ", this EventLoop at :" << this
                  << ". tid = " << m_tid;
        ::abort();
    }
    loopInThisThread = this;
    m_eventChannel.setErrorCallback(detail::defaultEventErrorHandler);
    m_eventChannel.setReadCallback([this](){
        this->handleFunctors();
    });
    m_eventChannel.enbaleRead();
}

EventLoop::~EventLoop() {
    assert(loopInThisThread != nullptr);
    loopInThisThread = nullptr;
    m_eventChannel.disableAll();
    removeChannel(m_eventChannel);
    ::close(m_eventChannel.getFd());
}

bool EventLoop::isInLoopThread() {
    return m_tid == CurrectThread::getTid();
}

void EventLoop::assertInLoopThread() {
    if (CurrectThread::getTid() != m_tid)
        abortNotInLoopThread();
}

void EventLoop::loop() {
    assertInLoopThread();
    m_looping = true;
    while (m_looping) {
        int nActive = m_poller->poll();
        if (nActive > 0) {
            LOG_INFO << "EventLoop::loop() " << nActive << "fd active";
            auto activeChannels = m_poller->getActiveChannels(nActive);
            m_poller->handleEvents(activeChannels);
        }else {
            LOG_DEBUG << "EventLoop::loop() but nothing happend";
        }
    }
}

void EventLoop::updateChannel(Channel& channel) {
    assertInLoopThread();
    m_poller->updateChannel(channel);
}

void EventLoop::removeChannel(Channel& channel) {
    assertInLoopThread();
    m_poller->removeChannel(channel);
}

void EventLoop::runInLoop(const EventLoopFunctor_t& functor) {
    if (CurrectThread::getTid() == m_tid) {
        functor();
    } else {
        enqueueLoop(functor);
    }
}

void EventLoop::runInLoop(EventLoopFunctor_t&& functor) {
    if (CurrectThread::getTid() == m_tid) {
        functor();
    } else {
        enqueueLoop(std::move(functor));
    }
}

void EventLoop::runAt(Timestamp ts, const TimerCallback_t& cb) {
    runInLoop([this, ts, cb](){
        this->m_timerQueue.addTimer(cb, ts, 0.0);
    });
}

void EventLoop::runAt(Timestamp ts, TimerCallback_t&& cb) {
    // lambda 与移动语义，C++14
    runInLoop([cb = std::move(cb), this, ts](){
        this->m_timerQueue.addTimer(std::move(cb), ts, 0.0);
    });
}

void EventLoop::enqueueLoop(const EventLoopFunctor_t& functor) {
    // operation for m_functors, use mutex
    {
        std::unique_lock<std::mutex> lock(m_functorMutex);
        m_functors.push_back(functor);
    }
    // wakeup
    detail::writeEvnetFd(m_eventChannel.getFd());
}

void EventLoop::enqueueLoop(EventLoopFunctor_t&& functor) {
    {
        std::unique_lock<std::mutex> lock(m_functorMutex);
        m_functors.push_back(std::move(functor));
    }
    // wakeup
    detail::writeEvnetFd(m_eventChannel.getFd());
}

////////////////////////////////////////////////////////
/////    Private methods
////////////////////////////////////////////////////////


void EventLoop::abortNotInLoopThread() {
    LOG_ERROR << "EventLoop::abortNotInLoopThread(), "
              << "CurrectThraed::tid: " <<CurrectThread::getTid()
              << ", EventLoop::m_tid: "<< m_tid;
    ::abort();
}

void EventLoop::handleFunctors() {
    LOG_DEBUG << "EventLoop::handleFunctors() " 
              << m_functors.size() << " functor(s)";
    assertInLoopThread();
    detail::readEventFd(m_eventChannel.getFd());
    std::deque<EventLoopFunctor_t> newFunctors;

    // copy on write
    {
        std::unique_lock<std::mutex> lock(m_functorMutex);
        newFunctors.swap(m_functors);
    }

    for (auto& functor: newFunctors) {
        functor();
    }
}




} // namespace net

} // namespace toym
